package fenbushi;

import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

import java.io.File;
import java.util.List;

public class fenbushi {
    public static void main(String[] args) throws Exception {
        System.out.println("00000");
//        执行离线计算SateSet,创建一个运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//        注册需要缓存的文件
        env.registerCachedFile("D:\\code\\flink_demo\\src\\main\\java\\fenbushi\\fenbushi.text","localfile");
//        数据源


        DataSet<Integer> source = env.fromElements(1,2,3,4,5,6,7,8,9,10);
//        执行map

        source.map(new RichMapFunction<Integer, String>() {

            private String cachedata = "";
            // 初始化
            @Override
            public void open(Configuration parameters) throws Exception {

//                初始化的时候读取缓存文件
                File file = getRuntimeContext().getDistributedCache().getFile("localfile");
                List<String> list_ = FileUtils.readLines(file);
                System.out.println(list_);
                cachedata = list_.get(0);
                System.out.println("11111111111");
                super.open(parameters);
            }

            @Override
            public String map(Integer value) {
                return value+cachedata;
            }


        }).print();
        System.out.println("222222222222");
    }
}
